package com.sc.test;

import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.TextView;

import androidx.annotation.Nullable;
import androidx.appcompat.app.AppCompatActivity;

import com.sc.test.bean.Student;
import com.sc.test.utils.Util;

import java.util.List;

import butterknife.BindView;
import butterknife.ButterKnife;
import butterknife.OnClick;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by Shen Chao on 2017-12-28-028.
 */

public class RxjavaActivity extends AppCompatActivity {
    public static final String TAG = "RxjavaActivity";
    private Util util = new Util();
    @BindView(R.id.tv_rxjava_content)
    TextView tvRxjavaContent;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rxjava);
        ButterKnife.bind(this);

    }

    @OnClick({R.id.btn_rxjava1, R.id.btn_rxjava2, R.id.btn_rxjava3, R.id.btn_rxjava4, R.id.btn_rxjava5
            , R.id.btn_rxjava6, R.id.btn_rxjava7, R.id.btn_rxjava8})
    public void onViewClicked(View view) {
        switch (view.getId()) {
            case R.id.btn_rxjava1:
                Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onNext(3);
                        emitter.onNext(4);
                        emitter.onComplete();
                    }
                });
                Observable<String> observable1 = new Observable<String>() {
                    @Override
                    protected void subscribeActual(Observer<? super String> observer) {
                        observer.onNext("a");
                        observer.onNext("b");
                        observer.onNext("c");
                        observer.onComplete();
                    }
                };
                Observer<String> observer = new Observer<String>() {
                    //新加入方法，在订阅后，在发送数据前执行
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.d("item", s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d("error", "error");
                    }

                    @Override
                    public void onComplete() {
                        Log.d("Complete", "Complete");
                    }
                };
                Observer<Integer> observer1 = new Observer<Integer>() {
                    //新加入方法，在订阅后，在发送数据前执行
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d("item", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d("error", "error");
                    }

                    @Override
                    public void onComplete() {
                        Log.d("Complete", "Complete");
                    }
                };
                observable.subscribe(observer1);
                observable1.subscribe(observer);

                break;
            case R.id.btn_rxjava2:
                final StringBuffer sb = new StringBuffer();
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("hello");
                        emitter.onNext("world");
                        emitter.onNext("!");
                        emitter.onComplete();
                    }
                }).subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        sb.append(s);
                        sb.append("\n");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
                String result = sb.toString();
                Util.openNewActivity(RxjavaActivity.this, ResultActivity.class, result);
                break;
            case R.id.btn_rxjava3:
                util.clearString();
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("a");
                        util.addString("emitter a");
                        emitter.onNext("b");
                        util.addString("emitter b");
                        emitter.onNext("c");
                        util.addString("emitter c");
                        emitter.onComplete();
                        util.addString("emitter onComplete");
                        emitter.onNext("d");
                        util.addString("emitter d");
                    }
                }).subscribe(new Observer<String>() {
                    private Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        disposable = d;
                        util.addString("onSubscribe");
                    }

                    @Override
                    public void onNext(String s) {
                        util.addString("onNext " + s);
                        if ("b".equals(s)) {
                            util.addString("disposable");
                            disposable.dispose();
                            util.addString(disposable.isDisposed() + "");
                        }
                    }

                    @Override
                    public void onError(Throwable e) {
                        util.addString("onError");
                    }

                    @Override
                    public void onComplete() {
                        util.addString("onComplete");
                    }
                });
                Util.openNewActivity(RxjavaActivity.this, ResultActivity.class, util.finishString());
                break;
            case R.id.btn_rxjava4:
                util.clearString();
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("a");
                        util.addString("emitter a");
                        emitter.onNext("b");
                        util.addString("emitter b");
                        emitter.onNext("c");
                        util.addString("emitter c");
                        emitter.onComplete();
                        util.addString("emitter onComplete");
                        emitter.onNext("d");
                        util.addString("emitter d");
                    }
                }).subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        util.addString(s);
                    }
                });
                Util.openNewActivity(RxjavaActivity.this, ResultActivity.class, util.finishString());
                break;
            case R.id.btn_rxjava5:
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
                        emitter.onNext(1);
                        emitter.onComplete();
                    }
                }).subscribeOn(Schedulers.newThread())
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .doOnNext(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.d(TAG, "after observeon mainThread,current thread: " + Thread.currentThread().getName());
                            }
                        }).observeOn(Schedulers.io())
                        .subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.d(TAG, "after observeon io,current thread: " + Thread.currentThread().getName());
                            }
                        });
                break;
            case R.id.btn_rxjava8:
                /**
                 * doondoOnSubscribe 线程为后边指定的subscribeOn线程，未指定为main
                 *                   指定多个subscribeOn，只第一个生效
                 * doondoOnSubscribe 越在后边越先执行
                 doOnSubscribe after trampoline: main
                 doOnSubscribe after main: RxComputationThreadPool-1
                 doOnSubscribe after new Thread: RxCachedThreadScheduler-1
                 doOnSubscribe: RxNewThreadScheduler-1
                 * doonnext 线程为当前线程 上个运行的是doOnSubscribe 在newThread
                 doOnNext: RxNewThreadScheduler-1
                 map thread: RxNewThreadScheduler-1
                 doOnNext after new Thread: RxNewThreadScheduler-1
                 * observeOn 会改变以后的它线程
                 map after main: main
                 doOnNext after main: main
                 * observeOn 可以随时改变线程
                 doOnNext after trampoline: RxCachedThreadScheduler-2
                 onNext : RxCachedThreadScheduler-2
                 */
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onComplete();
                    }
                })
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                Log.d(TAG, "doOnSubscribe: " + Thread.currentThread().getName());
                            }
                        })
                        .doOnNext(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.d(TAG, "doOnNext: " + Thread.currentThread().getName());
                            }
                        })
                        .subscribeOn(Schedulers.newThread())
                        .subscribeOn(Schedulers.single())
                        .map(new Function<Integer, String>() {
                            @Override
                            public String apply(Integer integer) throws Exception {
                                Log.d(TAG, "map thread: " + Thread.currentThread().getName());
                                return integer + "";
                            }
                        })
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                Log.d(TAG, "doOnSubscribe after new Thread: " + Thread.currentThread().getName());
                            }
                        })
                        .doOnNext(new Consumer<String>() {
                            @Override
                            public void accept(String s) throws Exception {
                                Log.d(TAG, "doOnNext after new Thread: " + Thread.currentThread().getName());
                            }
                        })
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .map(new Function<String, String>() {
                            @Override
                            public String apply(String s) throws Exception {
                                Log.d(TAG, "map after main: " + Thread.currentThread().getName());
                                return s + "1";
                            }
                        })
                        .doOnNext(new Consumer<String>() {
                            @Override
                            public void accept(String s) throws Exception {
                                Log.d(TAG, "doOnNext after main: " + Thread.currentThread().getName());
                            }
                        })
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                Log.d(TAG, "doOnSubscribe after main: " + Thread.currentThread().getName());
                            }
                        })
                        .subscribeOn(Schedulers.computation())
                        .observeOn(Schedulers.io())
                        .doOnNext(new Consumer<String>() {
                            @Override
                            public void accept(String s) throws Exception {
                                Log.d(TAG, "doOnNext after trampoline: " + Thread.currentThread().getName());
                            }
                        })
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                Log.d(TAG, "doOnSubscribe after trampoline: " + Thread.currentThread().getName());
                            }
                        })
                        .subscribe(new Consumer<String>() {
                            @Override
                            public void accept(String s) throws Exception {
                                Log.d(TAG, "onNext : " + Thread.currentThread().getName());
                            }
                        });
                break;
            case R.id.btn_rxjava6:
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("zhangsan");
                    }
                }).subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .map(new Function<String, Student>() {
                            @Override
                            public Student apply(String s) throws Exception {
                                return Util.strToStudent(s);
                            }
                        }).subscribe(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        tvRxjavaContent.setText(student.getName());
                    }
                });
                break;
            case R.id.btn_rxjava7:
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("wangwu");
                    }
                }).subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .flatMap(new Function<String, ObservableSource<Student.Course>>() {
                            @Override
                            public ObservableSource<Student.Course> apply(String s) throws Exception {

                                List<Student.Course> courseList = Util.strToList(s);
                                return Observable.fromIterable((Iterable<Student.Course>) courseList);
                            }
                        }).subscribe(new Consumer<Student.Course>() {
                    @Override
                    public void accept(Student.Course course) throws Exception {
                        tvRxjavaContent.append(course.getName());
                        tvRxjavaContent.append("\n");
                    }
                });
                break;
        }
    }
}
